package com.study.flink.demo;

import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.util.Collector;

import com.study.flink.entity.does.HumidityDO;
import com.study.flink.entity.does.TemperatureDO;
import com.study.flink.source.HumiditySource;
import com.study.flink.source.TemperatureSource;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**
 * 测试两个source的connection操作。
 * 
 * @projectName: flink-demo
 * @author: yandong
 * @description:
 * @date: 2019年7月28日 下午3:18:28
 * @version:
 */
public class ConnectionDemo {

    public void run(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        KeyedStream<HumidityDO, Tuple> humidityKeyedStream = env.addSource(new HumiditySource())
            .keyBy("timestamp");
        KeyedStream<TemperatureDO, Tuple> temperatureKeyedStream = env
            .addSource(new TemperatureSource()).keyBy("timestamp");
        humidityKeyedStream.connect(temperatureKeyedStream).flatMap(new ConnectFunction())
            .writeAsText("/data/bigdata/forflink/connection_demo."
                + new SimpleDateFormat("YYMMddHHmmss").format(new Date()));
        try {
            env.execute();
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 
     * @projectName: flink-demo
     * @author: yandong
     * @description: 这个链接操作的state消耗的内存是疯狂的。
     * @date: 2019年7月28日 下午4:24:29
     * @version:
     */
    @Slf4j
    public static class ConnectFunction
        extends RichCoFlatMapFunction<HumidityDO, TemperatureDO, OutputObject> {
        private static final long serialVersionUID = 1L;
        private ValueState<OutputObject> states;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            states = getRuntimeContext()
                .getState(new ValueStateDescriptor<OutputObject>("states", OutputObject.class));
        }

        @Override
        public void flatMap1(HumidityDO value, Collector<OutputObject> out) throws Exception {
            log.info(">>> " + value.toString());
            OutputObject obj = states.value();

            if (null == obj) {
                obj = new OutputObject();
            }

            obj.setTimestamp(value.getTimestamp());
            obj.setHumidity(value.getHumidity());
            states.update(obj);
        }

        @Override
        public void flatMap2(TemperatureDO value, Collector<OutputObject> out) throws Exception {
            log.info(">>> " + value.toString());
            OutputObject obj = states.value();

            if (null == obj) {
                obj = new OutputObject();
            }

            obj.setTemperature(value.getTemperature());
            states.update(obj);
            out.collect(obj);
        }

    }

    @Data
    public static class OutputObject {
        private String timestamp;
        private String humidity;
        private String temperature;

        @Override
        public String toString() {
            return String.format("[%s, %s, %s]", timestamp, humidity, temperature);
        }
    }
}
